package com.bw.medical.realtime.dwd.app;

import com.bw.medical.realtime.common.base.BaseSQL;
import com.bw.medical.realtime.common.constant.Constant;
import com.bw.medical.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdUserRegister extends BaseSQL {
    public static void main(String[] args) {
        new DwdUserRegister().start(
                10015,
                4,
                "medical_dwd_user_user_register"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        readOdsDb(tabEnv,"medical_dwd_user_user_register");

        /**
         * 需要对数据 进行一个脱敏(保护用户隐私)
         */
        Table uuResult = tabEnv.sqlQuery("select\n" +
                "`data`['id'] `id` ,\n" +
                "`data`['create_time'] `register_time` ,\n" +
                "concat('*@',split_index(`data`['email'],'@',1)) `email` ,\n" +
                "concat(substr(`data`['telephone'],1,3),'****') `telephone` ,\n" +
                "`data`['username'] `username` \n" +
                "from \n" +
                "topic_db \n" +
                "where `table` = 'user'\n" +
                "and `type` = 'insert'");

        tabEnv.createTemporaryView("uu_table",uuResult);
        uuResult.execute().print();
        tabEnv.executeSql("create table " + Constant.DWD_USER_USER_REGISTER + "(\n" +
                "`id` string,\n" +
                "`register_time` string,\n" +
                "`email` string,\n" +
                "`telephone` string,\n" +
                "`username` string\n" +
                ") " + SQLUtil.getKafkaSinkSQL(Constant.DWD_USER_USER_REGISTER));

        uuResult.executeInsert(Constant.DWD_USER_USER_REGISTER);


    }
}
